package net.bwie.realtime.jtp.dwd.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class AdjustIsNewProcessFunction extends KeyedProcessFunction<String,String,String> {
    //step1.声明状态
    private transient ValueState<String> firstDateState=null;

    @Override
    public void open(Configuration parameters) throws Exception {
        firstDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstDate", String.class));
    }
    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        //1.获取状态
        String firstDate = firstDateState.value();
        //2.判断状态是否为空
        JSONObject jsonObject = JSON.parseObject(value);
        JSONObject isNewValue = jsonObject.getJSONObject("common");
        Long tsValue = jsonObject.getLong("ts");
        if ("1".equals(isNewValue)) {
            if (firstDate == null) {
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT));
            } else if (firstDate == null) {
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue - 86400000, DateTimeUtil.DATE_FORMAT));
            }
        }
        out.collect(jsonObject.toJSONString());

    }
}
